import redis.asyncio as redis
import asyncio  # Import asyncio for running the event loop

# 连接池配置
pool = redis.ConnectionPool(
    host='localhost',
    port=6379,
    max_connections=10,
)

# 创建一个异步函数来初始化Redis连接
async def create_redis_connection():
    return redis.Redis(
        host='localhost',
        port=6379,
        db=0,
        password='123456',
        decode_responses=True,
        connection_pool=pool
    )

async def testAsyncConnect():
    r = await create_redis_connection()
    # 设置键值对
    await r.set("async_key", "hello, redis")
    # 获取值
    value = await r.get("async_key")
    print(value)
    # 关闭连接
    await r.close()

async def testAsyncWithPool():
    # 异步连接池
    pool = redis.ConnectionPool.from_url(
        "redis://localhost",
        max_connections=10,
        decode_responses=True
    )
    r = redis.Redis(connection_pool=pool)
    await r.set("pool_key", "value")
    print(await r.get("pool_key"))

    await r.close()

async def testAsyncPipeline():
    r = await create_redis_connection()

    # 创建异步管道
    pipe = r.pipeline()
    pipe.set("counter", 0)
    pipe.incr("counter")
    pipe.incr("counter")

    # 提交命令
    results = await pipe.execute()
    print(results)

async def testAsyncMulti():
    r = await create_redis_connection()

    async with r.pipeline() as pipe:
        try:
            await pipe.watch("balance") # 监视键
            balance = await pipe.get("balance")
            new_balance = int(balance) - 20
            # 开启事务
            pipe.multi()
            pipe.set("balance", new_balance)
            pipe.incrby("savings", 20)
            # 提交事务
            await pipe.execute()
            print(await r.get("savings"))
        except redis.WatchError:
            print("事务失败, balance被其他客户端修改")

async def testAsyncPubSub():
    # 建立 Redis 连接
    r = await create_redis_connection()
    # 清空 Redis 数据库
    await r.flushall()

    async def publish_message():
        # 发布消息到 "news" 频道
        await r.publish("news", "Hello, Async")

    # 订阅 "news" 频道
    pubsub = r.pubsub()
    await pubsub.subscribe("news")

    # 发布消息
    await publish_message()

    # 监听消息
    async for message in pubsub.listen():
        if message['type'] == 'message':
            print(f"收到消息: {message['data'].decode()}")
            break  # 退出循环（如果只想接收一次消息）

    # 关闭连接
    pubsub.unsubscribe()
    await r.close()


async def main():
    # await testAsyncConnect()
    # await testAsyncWithPool()
    # await testAsyncPipeline()
    # await testAsyncMulti()
    await testAsyncPubSub()

if __name__ == "__main__":
    asyncio.run(main())